package test.cluster.net.listener;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

import test.domain.environment.node.Node;

public class MinaListener implements Listener {
	private Log log = LogFactory.getLog(this.getClass());
	private Node node;
	
	private IoAcceptor acceptor;
	
	@Override
	public Listener initialize(Node node) {
		if (node == null)
			throw new IllegalArgumentException("node is required.");
		
		this.node = node;
		return this;
	}
	
	@Override
	public void start() {
		if (node == null)
			throw new IllegalStateException("listener must be initialized.");
		
		// start listening requests from master node
		acceptor = new NioSocketAcceptor();
		
//		acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
		acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
		acceptor.setHandler(new ListenerSessionHandler());
		acceptor.getSessionConfig().setReadBufferSize(2048);
		
		log.info("start listening on port " + node.getPort());
		try {			
			acceptor.bind(new InetSocketAddress(node.getPort()));
		}
		catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void stop() {
		if (this.acceptor == null) return;
		
		log.info("stopping listener...");
		
		acceptor.unbind();
		acceptor.dispose();
		
		log.info("stopped listener.");
	}
	
	public static void main(String[] args) throws InterruptedException {
		Node node = new Node();
		node.setIp("127.0.0.1");
		node.setPort(9000);
		
		Listener listener = new MinaListener().initialize(node);
		listener.start();
		
//		Thread.sleep(10000);
//		listener.stop();
	}
}
